package com.shujia.flink.sink;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

public class Demo3MySqlSInk {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> wordsDS = env.socketTextStream("master", 8888);

        //统计单词的数量
        DataStream<Tuple2<String, Integer>> countDS = wordsDS
                .map(word -> Tuple2.of(word, 1), Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(kv -> kv.f0)
                .sum(1);

        //将统计结果保存到数据
        countDS.addSink(new MySQlSink());

        env.execute();

    }
}

/**
 * 自定义sink将数据保存到mysql
 * RichSinkFunction:多了open和close方法，用于打开和关闭连接
 * SinkFunction
 */
class MySQlSink extends RichSinkFunction<Tuple2<String, Integer>> {
    Connection con;
    PreparedStatement stat;

    /**
     * invoke方法每一条数据执行一次
     */
    @Override
    public void invoke(Tuple2<String, Integer> kv, Context context) throws Exception {
        stat.setString(1, kv.f0);
        stat.setInt(2, kv.f1);

        //之哦行sql
        stat.execute();
    }

    /**
     * open方法会在任务启动的是偶，每一个task中执行一次
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        System.out.println("创建数据库连接");
        //1、加载启动
        Class.forName("com.mysql.jdbc.Driver");
        //2、创建数据库连接
        con = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata29", "root", "123456");
        //3、编写保存数据的sql
        //replace into 替换插入，如果没有就插入，如果有就更新，表需要有主键
        stat = con.prepareStatement("replace into word_count values(?,?)");
    }

    /**
     * close方法会在任务取消的是偶，每一个task中执行一次
     */
    @Override
    public void close() throws Exception {
        //4、关闭数据库连接
        stat.close();
        con.close();
    }


}